package com.shujia.action.base

import com.shujia.common.SparkMain
import org.apache.spark.sql.{DataFrame, SparkSession}

object UserActionBaseIptv extends SparkMain {

  override def run(spark: SparkSession): Unit = {
    //创建对象
    val userActionBaseIptv = UserActionBaseIptv(spark, day)

    //按照顺序调用方法
    userActionBaseIptv.clear()
    userActionBaseIptv.process()
    userActionBaseIptv.save()
  }


  //不适用new创建对象
  def apply(spark: SparkSession, day: String): UserActionBaseIptv = new UserActionBaseIptv(spark, day)
}

class UserActionBaseIptv(spark: SparkSession, day: String) extends UserActionBase {
  override def clear(): Unit = {
  }

  override def process(): Unit = {

    import spark.implicits._
    import org.apache.spark.sql.functions._
    /**
      * 读取精准搜索数据
      *
      */
    val iptvDF: DataFrame = spark
      .read
      .format("json")
      .load(Constants.USER_ACTION_IPTV_PATH + day)


    //处理数据
    df = iptvDF
      //过滤数据
      .where(trim($"stbid") =!= "")
      //将stbid和cid合并得到一个新的id
      .withColumn("uid", upper(md5(concat($"stbid", $"cid"))))
      .select($"uid", $"itemId", $"itemName", $"utctime", $"processDate")

  }

  override def save(): Unit = {
    save(Constants.USER_ACTION_BASE_IPTV_PATH, day)
  }

  override def load(): DataFrame = {
    load(spark, Constants.USER_ACTION_BASE_IPTV_PATH, day)
  }
}
